package hadoop.hdfs.demo;

import hadoop.entity.TestBean;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.db.DBConfiguration;
import org.apache.hadoop.mapreduce.lib.db.DBInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import hadoop.utils.FasterJsonTool;

import java.io.IOException;

/**
 * @author qiss
 * @create 2018-03-28 11:46
 **/
public class MysqlToHdfs {

    public static class DataAccessMap extends Mapper<LongWritable, TestBean,Text,Text>{
        @Override
        protected void map(LongWritable key, TestBean value, Context context) throws IOException, InterruptedException {
            context.write(new Text(value.getId().toString()),new Text(FasterJsonTool.writeValueAsString(value)));
        }
    }


    public static void main(String[] args) throws Exception{
        Configuration conf = new Configuration();
        DBConfiguration.configureDB(conf,"com.mysql.jdbc.Driver", "jdbc:mysql://20.0.0.3:3306/agent?useUnicode=true&characterEncoding=utf8", "root", "Manyi@123");
        Job job = new Job(conf, "test mysql connection");
        job.setJarByClass(MysqlToHdfs.class);
        job.setMapperClass(DataAccessMap.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        String[] fields= {"id","action_type","value","memo"};
        FileOutputFormat.setOutputPath(job,new Path("hdfs:hadoop-namenode1:9000/wordcount/testData"));
        String sql = "select * from test_iw_agent_coin_dict";
        DBInputFormat.setInput(job,TestBean.class,"test_iw_agent_coin_dict",sql,"id",fields);
        System.exit(job.waitForCompletion(true)? 0 : 1);
    }
}
